#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据生成器
业务含义：生成演示数据，用于学习数据仓库开发过程
支持生成资源元数据、接口调用日志、下载日志、用户申请记录等
"""

import random
import time
import json
import requests
from datetime import datetime, timedelta
from kafka import KafkaProducer
import mysql.connector
import clickhouse_driver

class DataGenerator:
    def __init__(self):
        self.mysql_conn = None
        self.clickhouse_conn = None
        self.kafka_producer = None
        self.setup_connections()
        
        # 演示数据配置
        self.departments = [
            "省统计局", "省民政厅", "省教育厅", "省卫健委", 
            "省交通厅", "省环保厅", "省人社厅", "省住建厅", "省金融办"
        ]
        
        self.resource_types = ["API", "DB", "EXCEL"]
        
        self.api_urls = [
            "http://api.example.com/enterprise/query",
            "http://api.example.com/traffic/realtime", 
            "http://api.example.com/environment/monitor",
            "http://api.example.com/finance/statistics",
            "http://api.example.com/education/data",
            "http://api.example.com/health/resources",
            "http://api.example.com/employment/stats",
            "http://api.example.com/realestate/data"
        ]
        
        self.file_names = [
            "广东省GDP统计数据.xlsx",
            "人口普查数据.xlsx", 
            "教育统计数据.xlsx",
            "医疗资源分布.xlsx",
            "交通流量数据.xlsx",
            "环境监测数据.xlsx",
            "就业统计信息.xlsx",
            "房地产数据.xlsx"
        ]
        
        self.user_types = ["企业用户", "科研机构", "政府部门", "个人用户"]
        
    def setup_connections(self):
        """建立数据库连接"""
        try:
            # MySQL连接
            self.mysql_conn = mysql.connector.connect(
                host='mysql',
                user='dw_user',
                password='dw_pass',
                database='data_warehouse',
                charset='utf8mb4'
            )
            
            # ClickHouse连接
            self.clickhouse_conn = clickhouse_driver.Client(
                host='clickhouse',
                port=9000,
                user='dw_user',
                password='dw_pass',
                database='data_warehouse'
            )
            
            # Kafka生产者
            self.kafka_producer = KafkaProducer(
                bootstrap_servers=['kafka:9092'],
                value_serializer=lambda x: json.dumps(x, ensure_ascii=False).encode('utf-8')
            )
            
            print("数据库连接建立成功")
            
        except Exception as e:
            print(f"数据库连接失败: {e}")
            time.sleep(10)  # 等待服务启动
            
    def generate_resource_metadata(self, count=50):
        """生成资源元数据"""
        print(f"开始生成{count}条资源元数据...")
        
        for i in range(count):
            resource = {
                'resource_name': f'资源_{i+1}_{random.choice(self.file_names)}',
                'resource_type': random.choice(self.resource_types),
                'department': random.choice(self.departments),
                'status': random.choice(['ENABLED', 'DISABLED']),
                'create_time': self.random_datetime(days_back=365),
                'update_time': self.random_datetime(days_back=30),
                'api_url': random.choice(self.api_urls) if random.random() > 0.5 else None,
                'table_name': f'table_{i+1}' if random.random() > 0.5 else None,
                'file_path': f'/data/file_{i+1}.xlsx' if random.random() > 0.5 else None,
                'description': f'这是第{i+1}个资源的描述信息'
            }
            
            # 插入MySQL
            self.insert_mysql_resource(resource)
            
            # 发送到Kafka
            self.kafka_producer.send('resource-metadata', resource)
            
        print("资源元数据生成完成")
        
    def generate_api_call_logs(self, count=1000):
        """生成接口调用日志"""
        print(f"开始生成{count}条接口调用日志...")
        
        for i in range(count):
            call_log = {
                'resource_id': random.randint(1, 50),
                'api_url': random.choice(self.api_urls),
                'call_time': self.random_datetime(days_back=7),
                'response_time': random.randint(50, 2000),
                'status_code': random.choices([200, 400, 500], weights=[0.9, 0.05, 0.05])[0],
                'ip_address': f'192.168.1.{random.randint(100, 254)}',
                'user_id': f'user_{random.randint(1, 100)}',
                'department': random.choice(self.departments),
                'request_params': json.dumps({'query': f'param_{i}'}),
                'response_data': json.dumps({'result': 'success'})
            }
            
            # 插入ClickHouse
            self.insert_clickhouse_api_log(call_log)
            
            # 发送到Kafka
            self.kafka_producer.send('api-call-log', call_log)
            
        print("接口调用日志生成完成")
        
    def generate_download_logs(self, count=500):
        """生成下载日志"""
        print(f"开始生成{count}条下载日志...")
        
        for i in range(count):
            download_log = {
                'resource_id': random.randint(1, 50),
                'file_name': random.choice(self.file_names),
                'download_time': self.random_datetime(days_back=7),
                'ip_address': f'192.168.1.{random.randint(100, 254)}',
                'user_id': f'user_{random.randint(1, 100)}',
                'file_size': random.randint(1024, 10*1024*1024),  # 1KB到10MB
                'department': random.choice(self.departments),
                'download_source': random.choice(['web', 'api', 'mobile']),
                'user_agent': random.choice([
                    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
                    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
                    'Python-requests/2.28.1'
                ])
            }
            
            # 插入ClickHouse
            self.insert_clickhouse_download_log(download_log)
            
            # 发送到Kafka
            self.kafka_producer.send('download-log', download_log)
            
        print("下载日志生成完成")
        
    def generate_user_applications(self, count=200):
        """生成用户申请记录"""
        print(f"开始生成{count}条用户申请记录...")
        
        for i in range(count):
            application_time = self.random_datetime(days_back=30)
            approval_time = application_time + timedelta(hours=random.randint(1, 72))
            
            application = {
                'resource_id': random.randint(1, 50),
                'user_id': f'user_{random.randint(1, 100)}',
                'department': random.choice(self.departments),
                'application_time': application_time,
                'approval_time': approval_time if random.random() > 0.1 else None,
                'status': random.choices(['PENDING', 'APPROVED', 'REJECTED'], weights=[0.1, 0.8, 0.1])[0],
                'approval_duration': random.randint(1, 72) if random.random() > 0.1 else None,
                'approver_id': f'admin_{random.randint(1, 10)}' if random.random() > 0.1 else None,
                'application_reason': f'申请原因_{i+1}',
                'approval_reason': f'审批原因_{i+1}' if random.random() > 0.1 else None
            }
            
            # 插入MySQL
            self.insert_mysql_application(application)
            
            # 发送到Kafka
            self.kafka_producer.send('user-application', application)
            
        print("用户申请记录生成完成")
        
    def insert_mysql_resource(self, resource):
        """插入MySQL资源数据"""
        try:
            cursor = self.mysql_conn.cursor()
            sql = """
            INSERT INTO ods_resource_metadata 
            (resource_name, resource_type, department, status, create_time, update_time, 
             api_url, table_name, file_path, description)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
            """
            cursor.execute(sql, (
                resource['resource_name'], resource['resource_type'], resource['department'],
                resource['status'], resource['create_time'], resource['update_time'],
                resource['api_url'], resource['table_name'], resource['file_path'], resource['description']
            ))
            self.mysql_conn.commit()
            cursor.close()
        except Exception as e:
            print(f"插入MySQL资源数据失败: {e}")
            
    def insert_clickhouse_api_log(self, call_log):
        """插入ClickHouse接口调用日志"""
        try:
            self.clickhouse_conn.execute(
                "INSERT INTO ods_api_call_log_ch VALUES",
                [(
                    call_log['resource_id'], call_log['api_url'], call_log['call_time'],
                    call_log['response_time'], call_log['status_code'], call_log['ip_address'],
                    call_log['user_id'], call_log['department'], call_log['request_params'],
                    call_log['response_data']
                )]
            )
        except Exception as e:
            print(f"插入ClickHouse接口调用日志失败: {e}")
            
    def insert_clickhouse_download_log(self, download_log):
        """插入ClickHouse下载日志"""
        try:
            self.clickhouse_conn.execute(
                "INSERT INTO ods_download_log_ch VALUES",
                [(
                    download_log['resource_id'], download_log['file_name'], download_log['download_time'],
                    download_log['ip_address'], download_log['user_id'], download_log['file_size'],
                    download_log['department'], download_log['download_source'], download_log['user_agent']
                )]
            )
        except Exception as e:
            print(f"插入ClickHouse下载日志失败: {e}")
            
    def insert_mysql_application(self, application):
        """插入MySQL用户申请记录"""
        try:
            cursor = self.mysql_conn.cursor()
            sql = """
            INSERT INTO ods_user_application 
            (resource_id, user_id, department, application_time, approval_time, status,
             approval_duration, approver_id, application_reason, approval_reason)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
            """
            cursor.execute(sql, (
                application['resource_id'], application['user_id'], application['department'],
                application['application_time'], application['approval_time'], application['status'],
                application['approval_duration'], application['approver_id'], 
                application['application_reason'], application['approval_reason']
            ))
            self.mysql_conn.commit()
            cursor.close()
        except Exception as e:
            print(f"插入MySQL用户申请记录失败: {e}")
            
    def random_datetime(self, days_back=30):
        """生成随机时间"""
        now = datetime.now()
        random_days = random.randint(0, days_back)
        random_hours = random.randint(0, 23)
        random_minutes = random.randint(0, 59)
        random_seconds = random.randint(0, 59)
        
        return now - timedelta(
            days=random_days, 
            hours=random_hours, 
            minutes=random_minutes, 
            seconds=random_seconds
        )
        
    def generate_all_data(self):
        """生成所有演示数据"""
        print("开始生成演示数据...")
        
        # 生成资源元数据
        self.generate_resource_metadata(50)
        
        # 生成接口调用日志
        self.generate_api_call_logs(1000)
        
        # 生成下载日志
        self.generate_download_logs(500)
        
        # 生成用户申请记录
        self.generate_user_applications(200)
        
        print("演示数据生成完成！")
        
    def close_connections(self):
        """关闭数据库连接"""
        if self.mysql_conn:
            self.mysql_conn.close()
        if self.clickhouse_conn:
            self.clickhouse_conn.disconnect()
        if self.kafka_producer:
            self.kafka_producer.close()

if __name__ == "__main__":
    generator = DataGenerator()
    try:
        generator.generate_all_data()
    finally:
        generator.close_connections()
